package SparkGraphXInAction

import java.awt.Color
import java.awt.image.{BufferedImage, DataBufferInt}
import java.io.File
import javax.imageio.ImageIO

import org.apache.log4j.{Level, Logger}
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.graphx.{Edge, Graph}
import org.apache.spark.mllib.clustering.PowerIterationClustering

/**
  * Created by Administrator on 2017/4/28 0028.
  * 使用PIC算法来对图像进行分割。之前使用余弦夹角计算距离，发现相似度太高，现在使用其它距离算法。
  * 在TestPIC3中，选用欧氏距离吧。
  */
object TestPIC3 {

  def main(args: Array[String]): Unit = {
    // 屏蔽日志
    Logger.getLogger("org.apache.spark").setLevel(Level.WARN)
    Logger.getLogger("org.eclipse.jetty.server").setLevel(Level.OFF)

    //设置运行环境
    val conf = new SparkConf().setAppName("SimpleGraphX").setMaster("local")
    val sc = new SparkContext(conf)
    val im = ImageIO.read(new File("data/BSDS300/images/train/105053.jpg"))
    val ims = im.getScaledInstance(im.getWidth/8,im.getHeight/8,java.awt.Image.SCALE_AREA_AVERAGING)
    val width = ims.getWidth(null)
    val height = ims.getHeight(null)
    val bi = new BufferedImage(width, height, BufferedImage.TYPE_INT_RGB)
    bi.getGraphics.drawImage(ims,0,0,null)
    //r中的前三个元素：(-10725046,0) ,   (-10199475,1) , (-10330546,2)
    //(-10725046,0) 里面的-10725046代表某个像素，0代表index。
    val r = sc.makeRDD(bi.getData.getDataBuffer.asInstanceOf[DataBufferInt].getData).zipWithIndex.cache
    //我要展示出r.cartesian(r)之后，产生的rdd是怎样的。
    //r.cartesian(r)产生的前4个元素是：((-10725046,0),(-10725046,0))，    ((-10725046,0),(-10199475,1))，    ((-10725046,0),(-10330546,2))， ((-10725046,0),(-9673389,3))
    val g = Graph.fromEdges(r.cartesian(r).cache().map(x => {
      //toVec返回的是长度为3的元素类型为Double的Array
      def toVec(a:Tuple2[Int, Long])={
        val c = new Color(a._1)
        Array[Double](c.getRed, c.getGreen, c.getBlue)
      }
      //输入的u或v都是长度为3的元素类型为Double的Array
      //比如u是(1.0, 4.0, 3.5)  而 v 是 (3.0, 2.5, 1.4)
      def cosineSimilarity(u:Array[Double], v:Array[Double])={
        //d对应公式中的分母的操作
        val d = Math.sqrt(u.map(a => a*a).sum * v.map(a => a*a).sum)
        //u.zip(v)的出来的结果是什么呢？Array[(Double, Double)],
        //按照我们前面的u和v的举例，u.zip(v)的结果是Array((1.0,3.0), (4.0,2.5), (3.5,1.4))
        if (d == 0.0) 0.0 else u.zip(v).map(a => a._1*a._2).sum/d
      }

      //定义欧氏距离计算公式
      def euclideanSimilarity(u:Array[Double], v:Array[Double])={
        Math.sqrt(u.zip(v).map(a =>(a._1-a._2)*(a._1-a._2)).sum)
      }

      //这个Edge中有三个元素，前两个元素都是像素点(顶点ID)的编号（由zipWithIndex创建）,第三个元素是两个像素点的“距离”（余弦相似度）来作为边的权重。
      Edge(x._1._2, x._2._2, euclideanSimilarity(toVec(x._1),toVec(x._2)))
    }), 0.0)
//    var count = 0
//    for(uuu <- g.triplets.collect){
//      println("count : "+count)
//      println(uuu)
//      count = count + 1
//    }
    //生成PIC模型。
    val m = new PowerIterationClustering().setK(2).run(g)
    //    println(m.asInstanceOf[AnyRef].getClass)
    //    println(m.assignments.asInstanceOf[AnyRef].getClass)
    //    println("elements of m.assignments")
    //    val list_ass = m.assignments.collect
    //    for(ass <- list_ass){
    //      val asscluster = ass.cluster
    //      val assid = ass.id
    //      println(ass.cluster.asInstanceOf[AnyRef].getClass)
    //      println("ass.id, ass.cluster : "+assid+", "+asscluster)
    //      if(asscluster == 1){
    //        println("========================================================")
    //        Thread.sleep(3000)
    //      }
    //    }
    //    println("whites : ")
    //    val list_white = m.assignments.filter(x => x.cluster == 0).take(20)
    println("blacks : ")
    val list_black = m.assignments.filter(x => x.cluster > 0).take(20)
    //colors规定了要生成的聚类图要用哪些颜色来渲染。
    val colors = Array(Color.white.getRGB, Color.black.getRGB)
    val bi2 = new BufferedImage(width, height, BufferedImage.TYPE_INT_RGB)
    m.assignments
      .map(a => (a.id/width, (a.id%width, colors(a.cluster))))
      .groupByKey
      .map(a => (a._1, a._2.toList.sortBy(_._1).map(_._2).toArray))
      .collect
      .foreach(x => bi2.setRGB(0,x._1.toInt, width, 1, x._2, 0, width))
    ImageIO.write(bi2, "PNG", new File("out.png"));
    sc.stop()
  }
}
